package com.dream.technology.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.AsyncConfigurer;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 *
 * @author chenqi
 * @description
 * @data 2020/9/3
 */
@Slf4j
@Configuration
@EnableAsync
public class ThreadPoolConfig implements AsyncConfigurer {

    @Autowired
    BeanFactory beanFactory;

    @Value("${spring.application.name}")
    private String prefix;

    /**
     * 计算密集型任务
     */
    @Value("${spring.threadPoolExecutor.calculation.coreSize: 3}")
    private int calculationCore;

    @Value("${spring.threadPoolExecutor.calculation.maxSize: 6}")
    private int calculationMaxSize;

    @Value("${spring.threadPoolExecutor.calculation.queueSize: 100}")
    private int calculationQueueSize;

    @Value("${spring.threadPoolExecutor.calculation.unit:SECONDS}")
    private String calculationUnit;

    @Value("${spring.threadPoolExecutor.calculation.keepAliveTime: 300}")
    private int calculationKeepAliveTime;


    /**
     * io密集型任务
     */

    @Value("${spring.threadPoolExecutor.Io.coreSize: 10}")
    private int ioCore;

    @Value("${spring.threadPoolExecutor.Io.maxSize: 300}")
    private int ioMaxSize;

    @Value("${spring.threadPoolExecutor.Io.queueSize: 1000}")
    private int ioQueueSize;

    @Value("${spring.threadPoolExecutor.Io.unit:SECONDS}")
    private String ioUnit;

    @Value("${spring.threadPoolExecutor.Io.keepAliveTime: 3000}")
    private int ioKeepAliveTime;


    /**
     * 定时任务池，一般配合@schdule使用
     *
     * @return
     */
    @Override
    @Bean("sfaThreadPool")
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setCorePoolSize(5);
        taskExecutor.setMaxPoolSize(20);
        taskExecutor.setQueueCapacity(100);
        taskExecutor.setKeepAliveSeconds(3000);
        taskExecutor.setThreadNamePrefix(prefix + "-");
        taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        taskExecutor.initialize();
        return taskExecutor;
    }

    /**
     * 计算型
     *
     * @return
     */

    @Bean("calculationThreadPool")
    public Executor getPool() {
        FeiHeThreadPoolExecutor calculationThreadPool = new FeiHeThreadPoolExecutor(calculationMaxSize,
                calculationMaxSize,
                calculationKeepAliveTime,
                TimeUnit.valueOf(calculationUnit.toUpperCase()),
                new ArrayBlockingQueue<>(calculationQueueSize),
                prefix);

        return calculationThreadPool;
    }
    /**
     * io型
     *
     * @return
     */

    @Bean("ioThreadPool")
    public Executor getIoPool() {
        FeiHeThreadPoolExecutor ioThreadPool = new FeiHeThreadPoolExecutor(ioMaxSize,
                ioMaxSize,
                ioKeepAliveTime,
                TimeUnit.valueOf(ioUnit.toUpperCase()),
                new ArrayBlockingQueue<>(ioQueueSize),
                prefix);
        return ioThreadPool;
    }


    public class FeiHeThreadPoolExecutor extends ThreadPoolExecutor {
        // 保存任务开始执行的时间，当任务结束时，用任务结束时间减去开始时间计算任务执行时间
        private ConcurrentHashMap<String, Long> startTimes;

        // 线程池名称，以业务名称命名，方便区分
        private String poolName;

        /**
         * 调用父类的构造方法，并初始化HashMap和线程池名称
         *
         * @param corePoolSize    线程池核心线程数
         * @param maximumPoolSize 线程池最大线程数
         * @param keepAliveTime   线程的最大空闲时间
         * @param unit            空闲时间的单位
         * @param workQueue       保存被提交任务的队列
         * @param poolName        线程池名称
         */
        public FeiHeThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue,
                                       String poolName) {
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, new EventThreadFactory(poolName));
            this.startTimes= new ConcurrentHashMap<>(maximumPoolSize);
            this.poolName=poolName;

        }

        /**
         * 任务执行之前，记录任务开始时间
         */
        @Override
        protected void beforeExecute(Thread t, Runnable r) {
            super.beforeExecute(t,r);
            startTimes.put(String.valueOf(r.hashCode()), System.currentTimeMillis());
        }



        /**
         * 任务执行之后，计算任务结束时间
         */
        @Override
        protected void afterExecute(Runnable r, Throwable t) {
            super.afterExecute(r,t);
            long diff = System.currentTimeMillis() - startTimes.remove(String.valueOf(r.hashCode()));
            // 统计任务耗时、初始线程数、核心线程数、正在执行的任务数量、已完成任务数量、任务总数、队列里缓存的任务数量、池中存在的最大线程数、最大允许的线程数、线程空闲时间、线程池是否关闭、线程池是否终止
            log.info(String.format(this.poolName
                            + "-pool-monitor: Duration: %d ms, PoolSize: %d, CorePoolSize: %d, Active: %d, Completed: %d, Task: %d, Queue: %d, LargestPoolSize: %d, MaximumPoolSize: %d,  KeepAliveTime: %d, isShutdown: %s, isTerminated: %s",
                    diff, this.getPoolSize(), this.getCorePoolSize(), this.getActiveCount(), this.getCompletedTaskCount(), this.getTaskCount(),
                    this.getQueue().size(), this.getLargestPoolSize(), this.getMaximumPoolSize(), this.getKeepAliveTime(TimeUnit.MILLISECONDS),
                    this.isShutdown(), this.isTerminated()));
        }


    }


    /**
     * 生成线程池所用的线程，只是改写了线程池默认的线程工厂，传入线程池名称，便于问题追踪
     */
    static class EventThreadFactory implements ThreadFactory {
        private static final AtomicInteger poolNumber = new AtomicInteger(1);
        private final ThreadGroup group;
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;

        /**
         * 初始化线程工厂
         *
         * @param poolName 线程池名称
         */
        EventThreadFactory(String poolName) {
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
            namePrefix = poolName + "-pool-" + poolNumber.getAndIncrement() + "-thread-";
        }

        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
            if (t.isDaemon())
                t.setDaemon(false);
            if (t.getPriority() != Thread.NORM_PRIORITY)
                t.setPriority(Thread.NORM_PRIORITY);
            return t;
        }
    }


}
